package com.foreveross.taskservice.engine.worker;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import com.foreveross.taskservice.common.CacheDataSpace;
import com.foreveross.taskservice.common.model.AbstractTaskModel;
import com.foreveross.taskservice.common.model.TaskAction;
import com.foreveross.taskservice.common.mq.QueueNameBuild;
import com.foreveross.taskservice.common.mq.destination.DestinationFactory;
import com.foreveross.taskservice.infra.cfg.PropertyUtils;
import com.foreveross.taskservice.infra.cfg.taskservice.TaskServicePropertiesLoader;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
/**
 * 将MQ中的任务回滚Worker
 * @author Administrator
 *
 */
public class TaskRollbackFromMQWorker extends AbstractTaskStatefulJobWorker{
	
	@Override
	public void execute(JobExecutionContext context)
			throws JobExecutionException {
		long stime = System.currentTimeMillis();
		log.info("任务回滚TaskRollbackFromMQWorker开始........");
		List<AbstractTaskModel> models = null;
		try{
			int size=PropertyUtils.getIntProperty("taskservice.mq.rollback.maxsize", TaskServicePropertiesLoader.FILE_NAME, 10);
			models=getRollbackTaskFromMQApplication().
					rolls(DestinationFactory.getInstance().getRollbackDestination(CacheDataSpace.AirFreightSpace()),size);
			log.info(String.format("任务回滚TaskRollbackFromMQWorker获得的任务数：%s........", models.size()));
			if(models!=null && !models.isEmpty()){
				List<List<AbstractTaskModel>> arange=arange(models);
					for(List<AbstractTaskModel> mList:arange){
						for(AbstractTaskModel t : mList){
							t.addAction(TaskAction.getRollbackAction());
						}
						int rollSize = getTaskCacheStorage().rollbacks(Sets.newLinkedHashSet(mList));
						log.info(String.format("正确回收任务数量[%s]", rollSize));
					}
			}
		}catch(Exception e){
			log.error(ExceptionUtils.getFullStackTrace(e));
		}finally{
			models = null;
			log.info("任务回滚TaskRollbackFromMQWorker结束....耗时ms:"+(System.currentTimeMillis()-stime));
		}
	}
	
	/**
	 * 整理任务
	 * 这里的目的是将同渠道同类型的任务放入同个list,一次性存入
	 * 这样可以避免多锁的情况
	 * 
	 * @param models
	 * @return
	 */
	private List<List<AbstractTaskModel>> arange(Collection<AbstractTaskModel> models){
		List<List<AbstractTaskModel>> result=Lists.newArrayList();
		Map<String,List<AbstractTaskModel>> map=new HashMap<String,List<AbstractTaskModel>>();
		CacheDataSpace space;
		String queueName;
		for(AbstractTaskModel m:models){
			space=new CacheDataSpace(m.getCreateTime(), m.getTaskType());
			queueName=QueueNameBuild.buildTaskQueueName(space, m.getChannel().getHbaseKey());
			if(!map.containsKey(queueName)){
				map.put(queueName, new ArrayList<AbstractTaskModel>());
			}
			map.get(queueName).add(m);
		}
		for(Entry<String, List<AbstractTaskModel>> en:map.entrySet()){
			result.add(en.getValue());
		}
		map.clear();
		return result;
	}

}
